package com.atguigu.flink.tableapi;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.OverWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2023/2/6
 */
public class Demo11_OverWindow
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTs());


        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);


        /*
                只选取流中数据的某些属性，组成Table
         */
        Table table = tableEnvironment.fromDataStream(ds, $("id"),$("ts"),$("vc"),
            $("pt").proctime(),$("et").rowtime()
        );

        /*
            OverWindow:    使用Over类定义
                            select xxx, sum(vc) over( partition by xx order by xxx [window clause]  )

                            window clause:  range|rows  between 上边界(xxx preceding)  and  下边界(xxx following | current row)

                 range: 以时间范围作为落入窗口的依据。只要两个数据的时间范围相等，进入同一个窗口
                 rows:  以行数作为落入窗口的依据。两个数据的时间范围相等，但是处于不同行，不会进入同一个窗口

                 只有水印超过当前数据的eventtime，才会触发运算！
         */

        //基于rows  范围是上无穷 到 当前行
        OverWindow w1 = Over.partitionBy($("id")).orderBy($("et")).preceding(UNBOUNDED_ROW).following(CURRENT_ROW).as("w");

        //基于rows  范围是 前2行 到 当前行
        OverWindow w2 = Over.partitionBy($("id")).orderBy($("et")).preceding(rowInterval(2l)).following(CURRENT_ROW).as("w");

        //和离线计算不同的地方   范围是 前2行 到 后2行  OVER RANGE FOLLOWING windows are not supported yet. 窗口的下限最多到当前行！ 错误示范
        //OverWindow w3 = Over.partitionBy($("id")).orderBy($("et")).preceding(rowInterval(2l)).following(rowInterval(2l)).as("w");

        //基于range  范围是上无穷 到 当前时间
        OverWindow w4 = Over.partitionBy($("id")).orderBy($("et")).preceding(UNBOUNDED_RANGE).following(CURRENT_RANGE).as("w");

        // 基于range  范围2S之前 到 当前时间  lit(): 用于构造一个字面量(字面上就能看出变量值的变量)  不支持定义下限为超过当前时间范围
        OverWindow w5 = Over.partitionBy($("id")).orderBy($("et")).preceding(lit(2).seconds()).following(CURRENT_RANGE).as("w");


        table.window(w5)
             .select($("id"),$("et"),$("vc"),$("vc").sum().over($("w")))
             .execute()
             .print();


    }
}
